package com.yjzx.util.binlog.core.listener;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import com.yjzx.util.binlog.core.config.BinLogBean;
import com.yjzx.util.binlog.core.constants.BaseConstant;
import com.yjzx.util.binlog.core.constants.MessageConstant;
import com.yjzx.util.binlog.core.entiry.*;
import com.yjzx.util.binlog.core.entiry.recordIndex.BinaryLogClientIndex;
import com.yjzx.util.binlog.core.executor.BinLogExecutor;
import com.yjzx.util.binlog.core.util.BinLogUtils;
import com.yjzx.util.binlog.core.util.CollectionUtil;
import com.yjzx.util.binlog.core.util.EmptyUtils;
import com.yjzx.util.binlog.core.util.StringUtils;
import lombok.extern.slf4j.Slf4j;
import net.sf.jsqlparser.JSQLParserException;
import net.sf.jsqlparser.parser.CCJSqlParserUtil;
import net.sf.jsqlparser.schema.Table;
import net.sf.jsqlparser.statement.Statement;
import net.sf.jsqlparser.statement.UnsupportedStatement;
import net.sf.jsqlparser.statement.alter.Alter;
import net.sf.jsqlparser.statement.alter.AlterExpression;
import net.sf.jsqlparser.statement.alter.RenameTableStatement;
import net.sf.jsqlparser.statement.create.table.ColumnDefinition;
import net.sf.jsqlparser.statement.create.table.CreateTable;
import net.sf.jsqlparser.statement.drop.Drop;

import java.io.IOException;
import java.io.Serializable;
import java.sql.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

/**
 * @author yjzx
 * @date 2024/8/15
 * @description mysql binlog监听器
 */
@Slf4j
public class MysqlBinLogListener implements BinaryLogClient.EventListener {
    /**
     * mysql的监听端
     */
    private final BinaryLogClient parseClient;
    /**
     * 消费者线程池
     */
    private final ExecutorService consumer;
    /**
     * database配置信息
     */
    private final BinLogBean properties;

    /**
     * 存放每张数据表对应的listener
     */
    private final Map<String, Set<BinLogExecutor>> listeners;
    /**
     * 数据库-表名 ：{字段名:列索引，列名，类型}
     */
    private final Map<String, ColumData[]> dbTableCols;
    /**
     * 数据库连接(使用后关闭)
     */
    private Connection connection;
    private transient volatile String database;
    private transient volatile String table;
    /**
     * 当前操作的表名
     */
    private transient volatile String dbTable;

    static class MyThreadFactory implements ThreadFactory {
        private final AtomicLong count = new AtomicLong(1);

        @Override
        public Thread newThread(Runnable r) {
            return new Thread(r, BaseConstant.BINLOG_DEFAULT_POOL_PREFIX + count.getAndIncrement());
        }
    }

    /**
     * 监听器初始化
     *
     * @param properties
     */
    public MysqlBinLogListener(BinLogBean properties) {
        this.properties = properties;
        Conf conf = properties.getDatasource();
        BinaryLogClient client = new BinaryLogClient(conf.getHost(), conf.getPort(), conf.getUsername(), conf.getPassword());
        EventDeserializer eventDeserializer = new EventDeserializer();

//        eventDeserializer.setCompatibilityMode(//序列化
//                EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG,
//                EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY
//        );
        client.setEventDeserializer(eventDeserializer);

        BinaryLogClientIndex clientIndex = properties.getReadBinLogIndex();
        if (clientIndex != null) {
            client.setBinlogFilename(clientIndex.getBinFileName());
            client.setBinlogPosition(clientIndex.getPosition());
        }

        this.parseClient = client;
        this.listeners = new HashMap<>();
        this.dbTableCols = new ConcurrentHashMap<>();
        int coreSize = properties.coreThreads;
        this.consumer = new ThreadPoolExecutor(coreSize, Math.max(properties.consumerThreads, coreSize), 30L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(100), new MyThreadFactory());
        // 保存当前注册的表的colum信息
        connection = sqlConnection(conf);
    }

    /**
     * mysql连接初始化
     *
     * @param conf 配置
     * @return 连接
     */
    public Connection sqlConnection(Conf conf) {
        Connection connection = null;
        try {
            Class.forName(conf.getClassname());
            connection = DriverManager.getConnection(BaseConstant.BINLOG_MYSQL_LINK_PREFIX + conf.getHost() + BaseConstant.BINLOG_MYSQL_LINK_SEPARATOR + conf.getPort() + BaseConstant.BINLOG_MYSQL_LINK_SUFFIX_DEFAULT, conf.getUsername(), conf.getPassword());
        } catch (SQLException e) {
            try {
                DriverManager.getConnection(BaseConstant.BINLOG_MYSQL_LINK_PREFIX + conf.getHost() + BaseConstant.BINLOG_MYSQL_LINK_SEPARATOR + conf.getPort() + BaseConstant.BINLOG_MYSQL_LINK_SUFFIX, conf.getUsername(), conf.getPassword());
            } catch (SQLException ex) {
                throw new RuntimeException(MessageConstant.BINLOG_MYSQL_LINK_ERROR + e.getMessage());
            }
        } catch (ClassNotFoundException e) {
            throw new RuntimeException(MessageConstant.BINLOG_MYSQL_VERSION_ERROR + e.getMessage());
        }
        return connection;
    }

    /**
     * 注册单监听
     *
     * @param db       数据库
     * @param table    操作表
     * @param listener 监听器
     * @throws Exception
     */
    public String regByTableListener(String db, String table, BinLogExecutor listener) {
        String dbTable = BinLogUtils.getdbTable(db, table);
        // 获取字段集合
        ColumData[] cols = getCols(db, table);
        // 保存字段信息
        dbTableCols.put(dbTable, cols);
        // 保存当前注册的listener
        listenersAdd(listener, dbTable);
        return dbTable;
    }

    /**
     * 注册监听器
     *
     * @param db       数据库
     * @param dbTables 当前数据库的指定表
     * @param listener 监听器
     * @return
     */
    public List<String> regListeners(String db, Set<String> dbTables, BinLogExecutor listener) {
        List<String> list = new ArrayList<>();
        if (dbTables.size() > 10) {
            // 获取字段集合
            Map<String, List<ColumData>> map = getTableColMap(db);
            for (Map.Entry<String, List<ColumData>> entry : map.entrySet()) {
                String table = entry.getKey();
                String dbTable = BinLogUtils.getdbTable(db, table);
                if (EmptyUtils.isEmpty(dbTables) ?
                        // 判断是否为空，为空则全部监听
                        listener.isAllowTable(dbTable) :
                        // 判断是否包含当前表，并且判断是否允许监听
                        dbTables.contains(dbTable) && listener.isAllowTable(dbTable)
                ) {
                    ColumData[] cols = entry.getValue().stream().sorted(Comparator.comparing(ColumData::getInx)).toArray(ColumData[]::new);
                    // 保存字段信息
                    dbTableCols.put(dbTable, cols);
                    // 保存当前注册的listener
                    listenersAdd(listener, dbTable);
                    list.add(dbTable);
                }

            }
        } else {
            for (String dbTableTemp : dbTables) {
                regByTableListener(db, dbTableTemp, listener);
            }
        }

        return list;
    }

    /**
     * 监听器单表注册
     *
     * @param listener 监听器
     * @param dbTable  数据库表
     */
    private void listenersAdd(BinLogExecutor listener, String dbTable) {
        Set<BinLogExecutor> set;
        if ((set = listeners.get(dbTable)) == null) {
            set = new HashSet<>();
            listeners.put(dbTable, set);
        }
        set.add(listener);
        if ((set = listeners.get("*")) == null) {
            set = new HashSet<>();
            listeners.put("*", set);

        }
        set.add(listener);
    }

    /**
     * 关闭数据库信息查询连接，开启监听binlog
     */
    public void finish() {
        try {
            if (connection != null) {
                connection.close();
                connection = null;
            }
        } catch (SQLException ignored) {

        }
        try {
            if (parseClient != null) {
                parseClient.registerEventListener(this);
                parseClient.connect();
            }
        } catch (IOException ignored) {
        }

    }

    /**
     * 获取指定数据库所有表，列
     *
     * @param db 数据库
     * @return 数据库表字段
     */
    private Map<String, List<ColumData>> getTableColMap(String db) {
        Map<String, List<ColumData>> columMap = Collections.emptyMap();
        try {
            // 执行sql
            String preSql = BaseConstant.BINLOG_MYSQL_DATABASE_SCHEMA_DATA_SQL;
            PreparedStatement ps = connection.prepareStatement(preSql);
            ps.setString(1, db);

            ResultSet rs = ps.executeQuery();
            ArrayList<Colum> list = new ArrayList<>(rs.getRow());
            while (rs.next()) {
                String schema = rs.getString(BaseConstant.BINLOG_MYSQL_TABLE_SCHEMA);
                String tableName = rs.getString(BaseConstant.BINLOG_MYSQL_TABLE_NAME);
                String column = rs.getString(BaseConstant.BINLOG_MYSQL_COLUMN_NAME);
                int idx = rs.getInt(BaseConstant.BINLOG_MYSQL_ORDINAL_POSITION);
                String dataType = rs.getString(BaseConstant.BINLOG_MYSQL_DATA_TYPE);
                if (column != null && idx >= 1) {
                    // sql的位置从1开始
                    list.add(new Colum(schema, tableName, idx - 1, column, dataType));
                }
            }
            columMap = CollectionUtil.toMapValueList(list, Colum::getTable, x -> new ColumData(x.inx, x.colName, x.dataType));

            rs.close();
            ps.close();
        } catch (SQLException e) {
            log.error(MessageConstant.BINLOG_MYSQL_LINK_DB_ERROR, db, e);
        }
        return columMap;
    }

    /**
     * 获取指定数据库表所有列
     */
    private ColumData[] getCols(String db, String table) {

        try (Connection sqlConnection = sqlConnection(this.properties.getDatasource())) {
            // 执行sql
            String preSql = BaseConstant.BINLOG_MYSQL_DATABASE_TABLE_DATA_SQL;
            PreparedStatement ps = sqlConnection.prepareStatement(preSql);
            ps.setString(1, db);
            ps.setString(2, table);
            ResultSet rs = ps.executeQuery();

            ColumData[] colums = new ColumData[rs.getRow()];
            while (rs.next()) {
                // String schema = rs.getString("TABLE_SCHEMA");
                // String tableName = rs.getString("TABLE_NAME");
                String column = rs.getString(BaseConstant.BINLOG_MYSQL_COLUMN_NAME);
                int idx = rs.getInt(BaseConstant.BINLOG_MYSQL_ORDINAL_POSITION);
                String dataType = rs.getString(BaseConstant.BINLOG_MYSQL_DATA_TYPE);
                if (column != null && idx >= 1) {
                    // sql的位置从1开始
                    colums[idx - 1] = new ColumData(idx - 1, column, dataType);
                }
            }
            rs.close();
            ps.close();
            return colums;
        } catch (SQLException e) {
            log.error(MessageConstant.BINLOG_MYSQL_LINK_DB_TABLE_ERROR, db, table, e);
        }
        return null;
    }

    /**
     * 监听处理
     *
     * @param event
     */
    @Override
    public void onEvent(Event event) {
        EventHeader header = event.getHeader();
        EventType eventType = header.getEventType();
        EventData data = event.getData();
        if (eventType == EventType.TABLE_MAP) {
            //获取当前操作的数据库和表
            TableMapEventData tableData = (TableMapEventData) data;
            this.dbTable = BinLogUtils.getdbTable((database = tableData.getDatabase()), (table = tableData.getTable()));
            return;
        }
        BinLogData item = null;
        QueryEventData queryData;
        if (eventType == EventType.QUERY && data instanceof QueryEventData && !"BEGIN".equalsIgnoreCase((queryData = (QueryEventData) data).getSql())) {
            //sql语句解析处理
            try {
                String database = BinLogUtils.nameToStr(queryData.getDatabase());
                Statement statement = CCJSqlParserUtil.parse(queryData.getSql());
                item = tableUpdate(statement, database, eventType, header);
            } catch (JSQLParserException e) {
                String sql = queryData.getSql();
                if (!sql.toUpperCase().contains("DROP DATABASE ")) {
                    String[] split = sql.split(" ");
                    String table = "";
                    if (split.length > 1) {
                        table = split[split.length - 1];
                    }
                    item = new BinLogDdl(this.database, table, eventType, header.getTimestamp(), header.getServerId(), sql);
                }
                log.error("转换sql异常：{};sql:{}", e.getMessage(), sql);
                //sql转换异常
            }
        }
        if (!listeners.containsKey(dbTable)) {
            return;
        }
        // 只处理添加删除更新三种操作
        if (EventType.isRowMutation(eventType)) {
            if (EventType.isWrite(eventType) && data instanceof WriteRowsEventData) {
                for (Serializable[] row : ((WriteRowsEventData) data).getRows()) {
                    item = dbTableCols.containsKey(dbTable) ? BinLogDml.itemFrom(database, table, row, dbTableCols.get(dbTable), eventType, header) : null;
                }
            } else if (EventType.isDelete(eventType) && data instanceof DeleteRowsEventData) {
                for (Serializable[] row : ((DeleteRowsEventData) data).getRows()) {
                    item = dbTableCols.containsKey(dbTable) ? BinLogDml.itemFrom(database, table, row, dbTableCols.get(dbTable), eventType, header) : null;
                }
            } else if (EventType.isUpdate(eventType) && data instanceof UpdateRowsEventData) {
                for (Map.Entry<Serializable[], Serializable[]> row : ((UpdateRowsEventData) data).getRows()) {
                    if (dbTableCols.containsKey(dbTable)) {
                        item = BinLogDml.itemFrom(database, table, row, dbTableCols.get(dbTable), eventType, header);
                    }
                }
            }
        }

        if (item != null) {
            final BinLogData finalItem = item;
            final String finalDbTable = dbTable;
            //充分利用线程池,为保证执行顺序，需此次执行结束才进行下次执行
            try {
                consumer.invokeAll(
                        listeners.get(finalDbTable).stream().map(binLogListener ->
                                (Callable<Object>) () -> {
                                    binLogListener.onEvent(finalItem);
                                    String binlogIndex = parseClient.getBinlogFilename() + BaseConstant.BINLOG_DEFAULT_INI_LINK_SEPARATOR + parseClient.getBinlogPosition();
                                    this.properties.getWriteBinLogIndex().forEach(x -> x.writeBinlogIndex(binlogIndex));
                                    return null;
                                }
                        ).collect(Collectors.toList()));
            } catch (InterruptedException ignored) {
            }
        }
    }

    /**
     * 表更新
     *
     * @param statement 监听信息
     * @param database  数据库
     * @param eventType 事件类型
     * @param header    事件头信息
     * @return 处理信息
     */
    private BinLogData tableUpdate(Statement statement, String database, EventType eventType, EventHeader header) {
        Result result = getDatabaseToTable(Collections.singletonList(this.table), database);
        if (statement instanceof Alter) {
            //监听表字段变更语句
            Alter alter = (Alter) statement;
            List<String> nameParts = alter.getTable().getNameParts();
            result = getDatabaseToTable(nameParts, database);
            if (this.dbTableCols.containsKey(result.dbTable)) {
                columnUpdate(alter.getAlterExpressions(), result.database, result.table);
            }
        } else if (statement instanceof RenameTableStatement) {
            //表名称变更语句
            for (Map.Entry<Table, Table> tableEntry : ((RenameTableStatement) statement).getTableNames()) {
                result = getDatabaseToTable(tableEntry.getKey().getNameParts(), database);
                ColumData[] columData;
                if ((columData = this.dbTableCols.get(result.dbTable)) != null) {
                    Result newTable = getDatabaseToTable(tableEntry.getKey().getNameParts(), database);
                    this.dbTableCols.put(newTable.dbTable, columData);
                    this.dbTableCols.remove(result.dbTable);
                }
            }
        } else if (statement instanceof Drop) {
            //删除表
            result = getDatabaseToTable(((Drop) statement).getName().getNameParts(), database);
            this.dbTableCols.remove(result.dbTable);
            this.listeners.remove(result.dbTable);
        } else if (statement instanceof CreateTable) {
            //添加表，但未注册表信息，需要监听当前表数据
            CreateTable createTable = (CreateTable) statement;
            result = getDatabaseToTable(createTable.getTable().getNameParts(), database);

            if (this.properties.getDbTable().containsKey(result.dbTable)) {
                Set<BinLogExecutor> set = new HashSet<>();
                //判断是否监听
                Result finalResult = result;
                boolean isAllTable = StringUtils.isEmpty(this.properties.getDbTable().get(result.dbTable));
                listeners.get("*").forEach(
                        binLogListener -> {
                            if ((isAllTable && binLogListener.isAllTable()) || binLogListener.isAllowTable(finalResult.dbTable)) {
                                set.add(binLogListener);
                            }
                        }
                );
                if (!set.isEmpty()) {
                    List<ColumnDefinition> list = createTable.getColumnDefinitions();
                    Value value = new Value(new ColumData[list != null ? list.size() : 0], 0);
                    if (list != null) {
                        for (ColumnDefinition definition : list) {
                            addColumn(definition, value);
                        }
                    }

                    listeners.put(result.dbTable, set);
                    this.dbTableCols.put(result.dbTable, value.toArray());
                }

            }

        } else if (statement instanceof UnsupportedStatement) {
            //添加数据库
        } else {
            log.warn("未定义的处理;sql:" + statement.toString());
        }
        return new BinLogDdl(result.database, result.table, eventType, header.getTimestamp(), header.getServerId(), statement.toString());
    }

    /**
     * 获取数据库及表信息
     *
     * @param nameParts 数据库及表信息
     * @param database  数据库
     * @return 返回
     */
    private Result getDatabaseToTable(List<String> nameParts, String database) {
        String table = this.table, dbTable = this.dbTable;
        if (nameParts != null) {
            if (nameParts.size() == 1) {
                dbTable = BinLogUtils.getdbTable(database, table = BinLogUtils.nameToStr(nameParts.get(0)));
            } else if (nameParts.size() == 2) {
                dbTable = BinLogUtils.getdbTable((database = BinLogUtils.nameToStr(nameParts.get(1))), (table = BinLogUtils.nameToStr(nameParts.get(0))));
            }
        }
        return new Result(database, table, dbTable);
    }

    private static class Result {
        public final String database;
        public final String table;
        public final String dbTable;

        public Result(String database, String table, String dbTable) {
            this.database = database;
            this.table = table;
            this.dbTable = dbTable;
        }
    }

    /**
     * 数据库表列操作类
     */
    private static class Value {
        private final ColumData[] data;
        private int len;

        public Value(ColumData[] data, int len) {
            this.data = data;
            this.len = len;
        }

        public ColumData[] toArray() {
            ColumData[] columData = new ColumData[len];
            for (int i = 0; i < len; i++) {
                (columData[i] = data[i]).inx = i;
            }
            return columData;
        }
    }

    /**
     * 数据库表字段更新
     * @param list 字段操作
     * @param database 数据库
     * @param table 表
     */
    private void columnUpdate(List<AlterExpression> list, String database, String table) {
        if (list == null || list.isEmpty()) {
            return;
        }
        String dbTable = BinLogUtils.getdbTable(database, table);
        ColumData[] arr = dbTableCols.get(dbTable);
        int length = arr.length;
        ColumData[] data = new ColumData[length + list.size()];
        System.arraycopy(arr, 0, data, 0, length);
        Value arrValue = new Value(data, length);

        for (AlterExpression expression : list) {
            switch (expression.getOperation()) {
                case ADD:
                    expression.getColDataTypeList().forEach(dataType -> addColumn(dataType, arrValue));
                    break;
                case MODIFY:
                    expression.getColDataTypeList().forEach(dataType -> modifyColumn(dataType, arrValue));
                    break;
                case DROP:
                case DROP_PRIMARY_KEY:
                case DROP_UNIQUE:
                case DROP_FOREIGN_KEY:
                    delColumn(expression, arrValue, dbTable);
                    break;
                case RENAME:
                case RENAME_TABLE:
                case RENAME_INDEX:
                case RENAME_KEY:
                case RENAME_CONSTRAINT:
                    changeColumn(expression, arrValue);
                    break;
                default:
                    throw new RuntimeException(dbTable + "Alter expression doesn't exist;" + expression.getOperation() + "sql :" + expression);
            }
        }
        dbTableCols.put(dbTable, arrValue.toArray());
    }

    private static void delColumn(AlterExpression expression, Value arrValue, String dbTable) {
        String col;
        if ((col = expression.getColumnName()) != null) {
            alterDel(arrValue, col);
        } else {
            throw new RuntimeException(dbTable + "DROP Alter expression doesn't exist;sql " + expression);
        }
    }

    private static ColumData alterDel(Value arrValue, String col) {
        col = BinLogUtils.nameToStr(col);
        ColumData[] data = arrValue.data;
        ColumData remove = null;
        for (int i = 0; i < arrValue.len; i++) {
            ColumData columData = data[i];
            if (columData.colName.equals(col)) {
                remove = columData;
                System.arraycopy(data, i + 1, data, i, arrValue.len - i - 1);
                arrValue.len -= 1;
                break;
            }
        }
        return remove;
    }

    private static void addColumn(ColumnDefinition dataType, Value arrValue) {
        boolean found = true;
        ColumData[] columData = arrValue.data;
        List<String> columnSpecs = dataType.getColumnSpecs();
        for (int i = 0; i < columnSpecs.size(); i++) {
            found &= alterAdd(dataType.getColumnName(), dataType.getColDataType().getDataType(), columnSpecs, 0, arrValue);
        }
        if (found) {
            columData[arrValue.len] = new ColumData(arrValue.len, BinLogUtils.nameToStr(dataType.getColumnName()), dataType.getColDataType().getDataType());
            arrValue.len++;
        }
    }

    private static boolean alterAdd(String newColumnName, String newDataType, List<String> columnSpecs, int i, Value arrValue) {
        newColumnName = BinLogUtils.nameToStr(newColumnName);
        ColumData[] columData = arrValue.data;
        boolean found = true;
        String columnName;
        switch (columnSpecs.get(i)) {
            case "after":
            case "AFTER":
                columnName = columnSpecs.get(i + 1);
                for (int i1 = 0; i1 < arrValue.len; i1++) {
                    if (columData[i1].colName.equals(columnName)) {
                        int start2 = i1 + 1;
                        System.arraycopy(columData, start2, columData, start2 + 1, arrValue.len - start2);
                        columData[start2] = new ColumData(start2, newColumnName, newDataType);
                        found = false;
                        arrValue.len++;
                        break;
                    }
                }
                break;
            case "first":
            case "FIRST":
                System.arraycopy(columData, 0, columData, 1, arrValue.len);
                columData[0] = new ColumData(arrValue.len, newColumnName, newDataType);
                arrValue.len++;
                found = false;
                break;
            case "before":
            case "BEFORE":
                columnName = columnSpecs.get(i + 1);
                for (int i1 = 0; i1 < arrValue.len; i1++) {
                    if (columData[i1].colName.equals(columnName)) {
                        System.arraycopy(columData, i1, columData, i1 + 1, arrValue.len - i1);
                        columData[i1] = new ColumData(i1, newColumnName, newDataType);
                        arrValue.len++;
                        found = false;
                        break;
                    }
                }
                break;
        }
        return found;
    }

    private static void modifyColumn(AlterExpression.ColumnDataType dataType, Value arrValue) {
        List<String> columnSpecs = dataType.getColumnSpecs();
        String columnName = BinLogUtils.nameToStr(dataType.getColumnName());
        alterModify(columnName, columnSpecs, arrValue);
    }

    /**
     * 字段修改
     * @param columnName2
     * @param columnSpecs
     * @param arrValue
     */
    private static void alterModify(String columnName2, List<String> columnSpecs, Value arrValue) {
        ColumData[] columData = arrValue.data;
        String order = "", columnName1 = null;
        int inx1 = 0, inx2 = 0;
        for (int i = 0; i < columnSpecs.size(); i++) {
            String col = columnSpecs.get(i);
            if ("after".equalsIgnoreCase(col)) {
                order = "after";
                columnName1 = BinLogUtils.nameToStr(columnSpecs.get(i + 1));
            } else if ("before".equalsIgnoreCase(col)) {
                order = "before";
                columnName1 = BinLogUtils.nameToStr(columnSpecs.get(i + 1));
            } else if ("first".equalsIgnoreCase(col)) {
                order = "first";
            }
        }
        for (int i = 0; i < arrValue.len; i++) {
            if (columData[i].colName.equals(columnName1)) {
                inx1 = i;
            }
            if (columData[i].colName.equals(columnName2)) {
                inx2 = i;
            }
        }
        ColumData columDatum = columData[inx2];
        if (inx2 > inx1 && ("after".equals(order) || "before".equals(order))) {
            columDatum = alterDel(arrValue, columDatum.colName);
            alterAdd(columDatum.colName, columDatum.dataType, Arrays.asList(order, columData[inx1].colName), 0, arrValue);
        } else if (inx2 < inx1 && "after".equals(order)) {
            alterAdd(columDatum.colName, columDatum.dataType, Arrays.asList(order, columData[inx1].colName), 0, arrValue);
            alterDel(arrValue, columDatum.colName);
        } else if (inx2 < inx1 && "before".equals(order)) {
            columDatum = alterDel(arrValue, columDatum.colName);
            alterAdd(columDatum.colName, columDatum.dataType, Arrays.asList(order, columData[inx1 - 1].colName), 0, arrValue);
        } else if ("first".equals(order)) {
            columDatum = alterDel(arrValue, columDatum.colName);
            alterAdd(columDatum.colName, columDatum.dataType, Arrays.asList(order, columData[inx1].colName), 0, arrValue);
        }

    }

    private static void changeColumn(AlterExpression expression, Value arrValue) {
        String columnOldName = BinLogUtils.nameToStr(expression.getColumnOldName());
        AlterExpression.ColumnDataType type = expression.getColDataTypeList().get(0);
        for (int i = 0; i < arrValue.len; i++) {
            ColumData datum = arrValue.data[i];
            if (columnOldName.equals(datum.colName)) {
                String columnName = BinLogUtils.nameToStr(type.getColumnName());
                datum.setColName(columnName);
                datum.setDataType(type.getColDataType().getDataType());
                List<String> columnSpecs = type.getColumnSpecs();
                if (columnSpecs != null && columnSpecs.size() > 0) {
                    alterModify(columnName, columnSpecs, arrValue);
                }
                break;
            }
        }
    }


}